package hanlen.mq.starter;

import hanlen.mq.starter.annotation.RedisQueueListener;
import hanlen.mq.starter.bean.MessageWrap;
import hanlen.mq.starter.constants.QueueConstants;
import hanlen.mq.starter.core.RedisQueueInterceptor;
import hanlen.mq.starter.core.RedissonService;
import hanlen.mq.starter.core.ThreadPoolExecutorService;
import hanlen.mq.starter.support.QueueUtil;
import hanlen.mq.starter.support.RedisQueueDlqSupport;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.slf4j.MDC;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

/**
 * @author weaponry MrYe
 * @program purchaser-assist-service
 * @description
 * @created 2020-09-25 15:21
 */
@Slf4j
public class RedisQueueBean<T> implements InitializingBean, ApplicationContextAware {

    private ApplicationContext context;
    private final T ref;
    private final List<Method> methods;
    private final Map<String, Thread> threadMap = new HashMap<>();

    RedisQueueBean(T ref, List<Method> methods) {
        this.ref = ref;
        this.methods = methods;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.context = applicationContext;
    }

    @Override
    public void afterPropertiesSet() {
        final RedissonService redissonService = context.getBean(RedissonService.class);
        final RedisQueueDlqSupport support = context.getBean(RedisQueueDlqSupport.class);
        final RedisQueueInterceptor interceptor = context.getBean(RedisQueueInterceptor.class);
        final ThreadPoolExecutorService executorService = context.getBean(ThreadPoolExecutorService.class);

        methods.forEach(m -> {
            final RedisQueueListener annotation = m.getAnnotation(RedisQueueListener.class);
            threadMap.putIfAbsent(annotation.name(), new Thread(() -> {
                final Thread thread = Thread.currentThread();
                boolean initFlag = false;
                final RBlockingQueue<MessageWrap> blockingQueue = redissonService.getBlockingQueue(annotation.name());
                while (true) {
                    if (!initFlag) {
                        initFlag = true;
                        log.info("register hanlen-redis-queue:{} thread-id:{} success", annotation.name(), thread.getId());
                    }
                    try {
                        TimeUnit.MILLISECONDS.sleep(2);
                        final MessageWrap msg = blockingQueue.take();
                        MDC.put(QueueConstants.MESSAGE_ID, msg.getMessageId());
                        final long backlogNums = redissonService.getAtomicLong(QueueConstants.getSizeKey(annotation.name())).decrementAndGet();
                        final Object payload = interceptor.preHandle(annotation.name(), (int) backlogNums, msg);
                        executorService.executor().execute(() -> {
                            try {
                                m.invoke(ref, payload);
                            } catch (Exception e) {
                                log.error("consumer hanlen-redis-queue:{} data:{} error:", annotation.name(), QueueUtil.toJsonString(payload), e);
                                support.addToDeadQueue(annotation, msg);
                            } finally {
                                MDC.remove(QueueConstants.MESSAGE_ID);
                            }
                        });
                    } catch (InterruptedException e) {
                        log.error("register hanlen-redis-queue:{} thread-id:{}", annotation.name(), thread.getId(), e);
                        break;
                    }
                }
            }));
        });
        threadMap.forEach((k, v) -> v.start());
    }

    void removeAllListener() {
        threadMap.forEach((k, v) -> v.interrupt());
    }

}
